package com.ruoyi.mqtt.endpoint;

import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@ServerEndpoint("/websocket/chat/{loginName}")
@Component
public class ChatWebSocketEndPoint {

    private String loginName;
    private Session session;

    //concurrent包的线程安全Set，用来存放每个客户端对应的Session对象。
    private static final Set<ChatWebSocketEndPoint> connections = new CopyOnWriteArraySet<>();
    private static AtomicInteger onlineNum = new AtomicInteger(0);//原子操作  整型 在线人数

    private static ScheduledExecutorService scheduledExecutorService = null;

    static {
        scheduledExecutorService = Executors.newScheduledThreadPool(10);
    }

    @PostConstruct
    public void beat(){
        scheduledExecutorService.scheduleAtFixedRate(()->{
            for (ChatWebSocketEndPoint client : connections){
                try {
                    client.session.getBasicRemote().sendText("[BEAT]");
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        },5, 5, TimeUnit.SECONDS);
    }

    /**
     * 当有ws 客户端连接进来，调用的方法
     * 处理连接事件
     * @param session
     */
    @OnOpen
    public void start(Session session, @PathParam("loginName")String loginName) {

        this.loginName = loginName;
        this.session = session;
        System.out.println("this:" + this);
        onlineNum.incrementAndGet();//++i  在线人数+1
        String msg = String.format("欢迎 %s(%s)加入，当前在线人数 %d",loginName,session.toString(),onlineNum.get());
        System.out.println(msg);

        connections.add(this);
        broadcast(msg);
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void end(Session session,@PathParam("loginName")String loginName) {
        connections.remove(this);
        onlineNum.decrementAndGet();//--i
        String msg = String.format("%s(%s) 退出，当前在线人数 %d",loginName,session.toString(),onlineNum.get());
        System.out.println(msg);
        broadcast(msg);
    }

    /**
     * 接收到消息的监听方法
     * client write    ------>   server  read
     * server write    ------>   client  read
     * @param message
     */
    @OnMessage
    public void incoming(Session session,String message,@PathParam("loginName")String loginName) {
        String msg = String.format("发送者%s(%s)发送消息： %s",loginName,session.toString(),message);
        System.out.println(msg);
        broadcast(msg);
    }

    /**
     * 出错了
     * @param t
     * @throws Throwable
     */
    @OnError
    public void onError(Throwable t) throws Throwable {
        System.out.println("onError");
        t.printStackTrace();
    }

    /**
     * 广播消息
     * @param msg
     */
    private void broadcast(String msg) {
        for (ChatWebSocketEndPoint client : connections) {
            try {
                synchronized (client) {
                    client.session.getBasicRemote().sendText(msg);
                }
            } catch (IOException e) {
                connections.remove(client);
                try {
                    client.session.close();
                } catch (IOException e1) {
                    // Ignore
                }
                String message = String.format("* %s %s",
                        client.loginName, "has been disconnected.");
                broadcast(message);
            }
        }
    }


}
